package com.yujinglin.OriginDes;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

public class OriginDesReducer extends Reducer<NullWritable, Text, Text, IntWritable> {

    private static final int TOP_N = 100; // 设置你想要的N值

    private TreeMap<Integer, Text> topNMap;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        topNMap = new TreeMap<>(Comparator.reverseOrder());
    }

    @Override
    protected void reduce(NullWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Map<Text, Integer> stringCounts = new HashMap<>();


        List<Text> valueList = new ArrayList<>();
        for (Text value : values) {
            valueList.add(new Text(value));
        }


        for (Text val : valueList) {

            stringCounts.put(val, stringCounts.getOrDefault(val, 0) + 1);
        }


        for (Text val : valueList) {
            if (topNMap.size() < TOP_N || (topNMap.size() == TOP_N && stringCounts.getOrDefault(val, 0) > topNMap.lastKey())) {

                if (topNMap.size() == TOP_N && stringCounts.getOrDefault(val, 0) > topNMap.lastKey()) {
                    topNMap.remove(topNMap.lastKey());
                }

                topNMap.put(stringCounts.getOrDefault(val, 0), val);
            }
        }


        for (Map.Entry<Integer, Text> entry : topNMap.entrySet()) {
            if (topNMap.size() <= TOP_N) {
                context.write(entry.getValue(), new IntWritable(entry.getKey()));
            } else {
                break;
            }
        }
    }
}
